import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
  * Action算子
  */
class ActionTest {
  /**
    * 1、reduceByKey是一个转换操作，reduce是一个Action算子
    * 2、reduceByKey是通过key来分组，然后每组聚合  reduce是一个数据集进行聚合
    * 3、reduceByKey是针对KV类型的   reduce是所有类型的数据
    */
  @Test
  //需求：最终生成("总价",price)
  /**
    * 注意点：
    * 1、curr 和 agg都是一个kv数据，而不是单纯的value
    * 2、返回的结果只有一个
    */
  def reduceTest(): Unit = {
    val conf = new SparkConf().setMaster("local[6]").setAppName("Action_Test")
    val sc = new SparkContext(conf)
    val rdd1 = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
    val result = rdd1.reduce((curr, agg) => ("总价", curr._2 + agg._2))
    println(result)
  }

  /**
    * foreach遍历出每一个元素
    * 注意：由于运行的时候是多线程的，因此打印出的结果可能和预期的顺序是不一样的
    */
  @Test
  def foreachTest(): Unit ={
    val conf = new SparkConf().setMaster("local[6]").setAppName("Action_Test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(Seq(1,2,3,4,5,6))
    rdd.foreach(item=>println(item))
  }

  /**
    * count求出元素的个数
    * countByKey求出key出现的次数:通过这个可以查看key对应的数据的数量，从而解决数据倾斜的问题
    * countByKey求出整个元素出现的次数
    */
  @Test
  def countTest(): Unit ={
    val conf = new SparkConf().setMaster("local[6]").setAppName("Action_Test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(Seq(("a",1),("b",1),("c",3),("a",4),("a",1)))
    println(rdd.count())
    println(rdd.countByKey())
    println(rdd.countByValue())
  }

  /**
    *take获取前几个,这个获取的是一个数组
    * first获取第一个
    * takeSample是抽样获取，第一个参数是释放放回，第二个参数是抽样抽取几个，返回的是数组
    */
  @Test
  def takeTest(): Unit ={
    val conf = new SparkConf().setMaster("local[6]").setAppName("Action_Test")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(Seq(1,2,3,4,5,6))
    rdd.take(3).foreach(println)
    println(rdd.first())
    rdd.takeSample(true,3).foreach(item=>println(item))
  }
}
